package com.dap.configsvrclient.disconf;

import java.io.InputStream;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Set;

import com.dap.configsvrclient.ConfigServerClient;
import com.dap.configsvrclient.RegistryObserver;
import com.dap.configsvrclient.disconf.rest.DisconfRestMgr;
import com.dap.configsvrclient.disconf.zk.ZkManager;
import com.dap.configsvrclient.disconf.zk.ZkNodeListener;


public class ConfigServerClientImpl implements ConfigServerClient, ZkNodeListener {
    Hashtable<String, Set<RegistryObserver>> configObservers = new Hashtable<>();
    Set<String> watched = new HashSet<>();


    @Override
    public boolean register(String path, String node, boolean isPersistent) {
        String nodeKey = ZkManager.getInstance().assembleZkNodeKey(path);
        boolean ret = true;
        if (!ZkManager.getInstance().isExists(nodeKey)) {
            ret = ZkManager.getInstance().create(nodeKey, null, isPersistent);
        }
        if ((!this.watched.contains(nodeKey)) && (ret)) {
            ret = ZkManager.getInstance().watch(nodeKey, this);
        }
        return ret;
    }


    @Override
    public boolean subscribe(String path, RegistryObserver observer) {
        Set<RegistryObserver> observers = this.configObservers.get(path);
        if (observers == null) {
            observers = new HashSet<>();
            this.configObservers.put(path, observers);
        }
        observers.add(observer);
        return register(path, null, true);
    }


    @Override
    public boolean isExist(String path) {
        String nodeKey = ZkManager.getInstance().assembleZkNodeKey(path);
        return ZkManager.getInstance().isExists(nodeKey);
    }


    @Override
    public InputStream getConfigContent(String configKey) {
        return DisconfRestMgr.getInstance().retrieveConfig(configKey);
    }


    @Override
    public ConfigServerClient.ConfigServerInfo getRegistryInfo() {
        ConfigServerClient.ConfigServerInfo info = new ConfigServerClient.ConfigServerInfo();
        info.registryHost = DisconfRestMgr.getInstance().retrieveZkHost();
        info.registryRootPath = DisconfRestMgr.getInstance().retrieveZkPrefix();
        return info;
    }


    @Override
    public void update(String configKey) {
        Set<RegistryObserver> observers = this.configObservers.get(configKey);
        if (observers != null) {
            for (RegistryObserver observer : observers) {
                observer.registryChanged(configKey);
            }
        }
    }
}
